package org.caw.util;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import java.util.Map.Entry;
import java.util.NavigableMap;

import org.apache.hadoop.hbase.HColumnDescriptor;
import org.apache.hadoop.hbase.HTableDescriptor;
import org.apache.hadoop.hbase.client.Delete;
import org.apache.hadoop.hbase.client.Get;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.util.Bytes;
import org.caw.domain.Family;
import org.caw.domain.Qualifier;
import org.caw.domain.Row;
import org.caw.domain.Version;
import org.caw.domain.descriptor.FamilyDescriptor;
import org.caw.domain.descriptor.TableDescriptor;

public class ConvertUtil {

	private static final String COLUMN_FAMILY_REGEX = "[\\S]+:[\\S]+";

	private static final String COLON = ":";

	public static Scan buildScan(String startKey, String endKey, List<String> columns, Integer maxversions, Long timestamp) {
		Scan scan = new Scan();

		if (startKey != null) {
			scan.setStartRow(Bytes.toBytes(startKey));
		}

		if (endKey != null) {
			scan.setStopRow(Bytes.toBytes(endKey));
		}

		if (timestamp != null) {
			scan.setTimeStamp(timestamp.longValue());
		} else {
			if (maxversions != null) {
				scan.setMaxVersions(maxversions.intValue());
			}
		}

		if (columns != null && !columns.isEmpty()) {
			for (String column : columns) {
				if (column.matches(COLUMN_FAMILY_REGEX)) {
					String[] data = column.split(COLON);
					scan.addColumn(Bytes.toBytes(data[0]), Bytes.toBytes(data[1]));
				} else {
					scan.addFamily(Bytes.toBytes(column));
				}
			}
		}

		return scan;
	}

	public static Get buildGet(String rowkey, List<String> columns, Integer maxversion, Long timestamp) throws IOException {
		Get get = new Get(Bytes.toBytes(rowkey));
		if (columns != null && !columns.isEmpty()) {
			for (String column : columns) {
				if (column.matches(COLUMN_FAMILY_REGEX)) {
					String[] data = column.split(COLON);
					get.addColumn(Bytes.toBytes(data[0]), Bytes.toBytes(data[1]));
				} else {
					get.addFamily(Bytes.toBytes(column));
				}
			}
		}

		get.setMaxVersions(maxversion == null ? 1 : maxversion);
		if (timestamp != null) {
			get.setTimeStamp(timestamp);
		}
		return get;
	}

	public static Delete buildDelete(String rowkey, List<String> columns, Long timestamp) {
		Delete delete = new Delete(Bytes.toBytes(rowkey));

		if (columns != null && !columns.isEmpty()) {
			for (String column : columns) {
				if (column.matches(COLUMN_FAMILY_REGEX)) {
					String[] data = column.split(COLON);
					delete.deleteColumn(Bytes.toBytes(data[0]), Bytes.toBytes(data[1]));
				} else {
					delete.deleteFamily(Bytes.toBytes(column));
				}
			}
		}
		delete.setTimestamp(timestamp != null ? timestamp : Long.MAX_VALUE);

		return delete;
	}

	public static Row convert(Result result) {
		Row row = new Row();
		row.setRowKey(Bytes.toString(result.getRow()));
		for (Entry<byte[], NavigableMap<byte[], NavigableMap<Long, byte[]>>> familyEntry : result.getMap().entrySet()) {

			if (row.getFamilies() == null) {
				row.setFamilies(new ArrayList<Family>());
			}

			Family family = new Family();
			family.setName(Bytes.toString(familyEntry.getKey()));
			family.setQualifiers(new ArrayList<Qualifier>());

			for (Entry<byte[], NavigableMap<Long, byte[]>> qualifierEntry : familyEntry.getValue().entrySet()) {
				Qualifier qualifier = new Qualifier();
				qualifier.setName(Bytes.toString(qualifierEntry.getKey()));

				qualifier.setVersions(new ArrayList<Version>());
				String latestValue = null;
				long timestamp = Long.MIN_VALUE;

				for (Entry<Long, byte[]> versionEntry : qualifierEntry.getValue().entrySet()) {
					Version version = new Version();
					version.setTimestamp(versionEntry.getKey());
					version.setValue(Bytes.toString(versionEntry.getValue()));
					qualifier.getVersions().add(version);

					// latest value
					if (timestamp < versionEntry.getKey()) {
						latestValue = Bytes.toString(versionEntry.getValue());
						timestamp = versionEntry.getKey();
					}
				}
				qualifier.setValue(latestValue);
				family.getQualifiers().add(qualifier);
			}

			row.getFamilies().add(family);
		}
		return row;
	}

	public static Put convert(Row row) {
		Put put = new Put(Bytes.toBytes(row.getRowKey()));
		for (Family family : row.getFamilies()) {
			for (Qualifier qualifier : family.getQualifiers()) {
				put.add(Bytes.toBytes(family.getName()), Bytes.toBytes(qualifier.getName()), Bytes.toBytes(qualifier.getValue()));
				// FIXME add versions
			}
		}
		return put;
	}

	public static TableDescriptor convert(HTableDescriptor desc, Boolean detailed) {
		TableDescriptor descriptor = new TableDescriptor();
		descriptor.setName(desc.getNameAsString());
		if (Boolean.TRUE.equals(detailed)) {
			descriptor.setFamilyDescriptors(new ArrayList<FamilyDescriptor>());
			for (HColumnDescriptor columnDesc : desc.getFamilies()) {
				FamilyDescriptor familyDesc = new FamilyDescriptor();
				familyDesc.setVersions(columnDesc.getMaxVersions());
				familyDesc.setBlockCache(columnDesc.isBlockCacheEnabled());
				familyDesc.setInMemory(columnDesc.isInMemory());
				familyDesc.setName(columnDesc.getNameAsString());
				familyDesc.setBlockSize(columnDesc.getBlocksize());
				familyDesc.setBloomFilterType(columnDesc.getBloomFilterType());
				familyDesc.setCompactionCompressionType(columnDesc.getCompactionCompressionType());
				familyDesc.setCompressionType(columnDesc.getCompressionType());
				familyDesc.setScope(columnDesc.getScope());
				familyDesc.setTtl(columnDesc.getTimeToLive());
				descriptor.getFamilyDescriptors().add(familyDesc);
			}
		}
		return descriptor;
	}

	public static HTableDescriptor convert(TableDescriptor desc) {
		HTableDescriptor descriptor = new HTableDescriptor();
		descriptor.setName(Bytes.toBytes(desc.getName()));
		if (desc.getFamilyDescriptors() != null) {

			for (FamilyDescriptor columnDesc : desc.getFamilyDescriptors()) {
				HColumnDescriptor familyDesc = new HColumnDescriptor(Bytes.toBytes(columnDesc.getName()));

				if (columnDesc.getBlockSize() != null) {
					familyDesc.setBlocksize(columnDesc.getBlockSize());
				}
				if (columnDesc.getVersions() != null) {
					familyDesc.setMaxVersions(columnDesc.getVersions().intValue());
				}
				if (columnDesc.getBlockCache() != null) {
					familyDesc.setBlockCacheEnabled(columnDesc.getBlockCache().booleanValue());
				}
				if (columnDesc.getInMemory() != null) {
					familyDesc.setInMemory(columnDesc.getInMemory().booleanValue());
				}
				if (columnDesc.getScope() != null) {
					familyDesc.setScope(columnDesc.getScope().intValue());
				}
				if (columnDesc.getTtl() != null) {
					familyDesc.setTimeToLive(columnDesc.getTtl().intValue());
				}
				if (columnDesc.getBloomFilterType() != null) {
					familyDesc.setBloomFilterType(columnDesc.getBloomFilterType());
				}
				if (columnDesc.getCompactionCompressionType() != null) {
					familyDesc.setCompactionCompressionType(columnDesc.getCompactionCompressionType());
				}
				if (columnDesc.getCompressionType() != null) {
					familyDesc.setCompressionType(columnDesc.getCompressionType());
				}
				descriptor.addFamily(familyDesc);
			}
		}
		return descriptor;
	}

	public static List<String> filterQualifiers(List<String> columns) {
		List<String> families = new ArrayList<String>();
		for (String column : columns) {
			String family = null;
			if (column.matches("[\\S]+:[\\S]+")) {
				family = column.split(":")[0];
			} else {
				family = column;
			}

			if (!families.contains(family)) {
				families.add(family);
			}
		}
		return families;
	}
}
